package com.zhouyu.fileSync.FSserver.registerServer;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.zhouyu.fileSync.FScommon.api.entities.MessageType;
import com.zhouyu.fileSync.FScommon.api.entities.Request;
import com.zhouyu.fileSync.FScommon.api.entities.Response;
import com.zhouyu.fileSync.FScommon.entities.FSClient;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.AttributeKey;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.io.IOException;
import java.lang.reflect.Method;
import java.net.SocketAddress;
import java.util.HashMap;

/**
 * Created by zhouyu@lightinthebox.com on 2019/8/14
 **/
public class RegisterServerHandler extends ChannelInboundHandlerAdapter {
    Logger logger = LogManager.getLogger(RegisterServerHandler.class);

    private final HashMap<String, Object> serviceMap;


    public RegisterServerHandler(HashMap<String, Object> serviceMap) {
        this.serviceMap = serviceMap;
    }


    @Override
    public void channelRead(io.netty.channel.ChannelHandlerContext ctx, java.lang.Object msg) throws java.lang.Exception {
        Request request = JSON.parseObject(msg.toString(), Request.class);
        if (request.getMessageType() == MessageType.METHOD) {
            logger.info("RPC客户端请求接口:" + request.getClassName() + "   方法名:" + request.getMethodName());
            Response response = new Response();
            response.setRequestId(request.getRequestId());
            try {
                if (request.getMethodName().equalsIgnoreCase("registerClient")){
                    //add client attribute to channel
                    AttributeKey<String> key=AttributeKey.valueOf("clientId");
                    FSClient fsc= JSONObject.toJavaObject((JSON) request.getParameters()[0],FSClient.class) ;
                    ctx.channel().attr(key).set(fsc.getClientName()+"@"+fsc.getClientPassword());
                    SocketAddress sa=ctx.channel().remoteAddress();
                    fsc.setClientHost(sa.toString().split(":")[0].substring(1));
                    request.getParameters()[0]=JSONObject.toJSON(fsc);
                }
                Object result = this.handler(request);
                response.setData(result);
            } catch (Throwable e) {
                e.printStackTrace();
                response.setCode(1);
                response.setErrorMsg(e.toString());
                logger.error("RPC Server handle request error", e);
            }
            ctx.writeAndFlush(response);
        } else if (request.getMessageType() == MessageType.HEARTBEAT) {
            logger.info("RPC客户端发送心跳信息");
        }
    }


    /**
     * 通过反射，执行本地方法
     *
     * @param request
     * @return
     * @throws Throwable
     */
    private Object handler(Request request) throws Throwable {
        String className = request.getClassName();
        Object serviceBean = serviceMap.get(className);
        if (serviceBean != null) {
            Class<?> serviceClass = serviceBean.getClass();
            String methodName = request.getMethodName();
            Class<?>[] parameterTypes = request.getParameterTypes();
            Object[] parameters = request.getParameters();

            Method method = serviceClass.getMethod(methodName, parameterTypes);
            method.setAccessible(true);
            return method.invoke(serviceBean, getParameters(parameterTypes, parameters));
        } else {
            throw new Exception("未找到服务接口,请检查配置!:" + className + "#" + request.getMethodName());
        }
    }

    /**
     * 获取参数列表
     *
     * @param parameterTypes
     * @param parameters
     * @return
     */
    private Object[] getParameters(Class<?>[] parameterTypes, Object[] parameters) {
        if (parameters == null || parameters.length == 0) {
            return parameters;
        } else {
            Object[] new_parameters = new Object[parameters.length];
            for (int i = 0; i < parameters.length; i++) {
               if (parameters[i] instanceof JSON){
                   new_parameters[i] = JSONObject.toJavaObject((JSON) parameters[i], parameterTypes[i]);
               }else{
                   new_parameters[i] = parameters[i];
               }
            }
            return new_parameters;
        }
    }


    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent event = (IdleStateEvent) evt;
            if (event.state() == IdleState.ALL_IDLE) {
                logger.info("no message had been received or send in the last 10s ,close the channel.{}", ctx.channel().remoteAddress());
                //clear discount client
                AttributeKey<String> key=AttributeKey.valueOf("clientId");
                RegisterServer.FSC_MAP.remove(ctx.channel().attr(key).get());
                ctx.channel().close();
            }
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }


    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        if (cause instanceof IOException){
            AttributeKey<String> key=AttributeKey.valueOf("clientId");
            RegisterServer.FSC_MAP.remove(ctx.channel().attr(key).get());
        }
        logger.info(cause.getMessage());
        ctx.close();
    }


}
